package cn.admin.configcenter.registry.common.task;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantLock;

public class TaskManager {

    private static final Logger log = LoggerFactory.getLogger(TaskManager.class);
    private static final long DEFAULT_SLEEP_TIME = 50L;
    protected final HashMap<String,Task> tasks;
    private final HashMap<String,TaskProcessor> processors;
    private final ReentrantLock lock;
    private final Thread processingThread;
    private final AtomicLong historyTasks;

    public TaskManager(String name) {
        this(name,32);
    }

    public TaskManager(String name,int initCapacity) {
        this.lock = new ReentrantLock();
        this.historyTasks = new AtomicLong(0L);
        this.tasks = new HashMap<>(initCapacity);
        this.processors = new HashMap<>();
        this.processingThread = new Thread(new ProcessRunnable());
        this.processingThread.setDaemon(true);
        this.processingThread.start();
    }

    public boolean isEmpty() {
        return this.tasks.isEmpty();
    }

    public int taskCount() {

        this.lock.lock();
        int count;
        try {
            count = this.tasks.size();
        } finally {
            this.lock.unlock();
        }
        return count;
    }

    public void addTask(Task newTask) {

        this.lock.lock();
        try {
            String taskId = newTask.getTaskId();
            Task existTask = this.tasks.get(taskId);
            if (existTask == null) {
                this.tasks.put(taskId,newTask);
            } else {
                newTask.merge(existTask);
                this.tasks.put(taskId,newTask);
            }
        } finally {
            this.lock.unlock();
        }

    }

    public AtomicLong getHistoryTasks() {
        return this.historyTasks;
    }

    public void addProcessor(String type, TaskProcessor taskProcessor) {
        this.lock.lock();

        try {
            this.processors.put(type, taskProcessor);
        } finally {
            this.lock.unlock();
        }

    }

    public void removeProcessor(String type) {
        this.lock.lock();

        try {
            this.processors.remove(type);
        } finally {
            this.lock.unlock();
        }

    }

    protected TaskProcessor getProcessor(Task task) {
        return this.processors.get(task.getType());
    }

    private Set<String> copyKeys() {
        this.lock.lock();
        Set<String> keys;
        try {
             keys = new HashSet<>(this.tasks.keySet());
        } finally {
            this.lock.unlock();
        }
        return keys;
    }

    private Task removeTask(String taskId) {
        this.lock.lock();
        Task task;
        try {
            task = this.tasks.get(taskId);
            if (task != null && task.shouldProcess()) {
                task = tasks.remove(taskId);
                return task;
            }
            task = null;
        } finally {
            this.lock.unlock();
        }
        return task;
    }

    private void processTasks() {

        Set<String> keys = this.copyKeys();

        for (String taskId : keys) {
            Task task = this.removeTask(taskId);
            if (task != null) {
                TaskProcessor processor = this.getProcessor(task);
                if (processor == null) {
                    log.error("########## processor not found for task, so discarded. " + task);
                } else {
                    try {
                        if (!processor.process(task)) {
                            this.addTask(task);
                        }
                    } catch (Exception e) {
                        log.error("Task execute error : " + e.toString(), e);
                    }
                }
            }
        }

    }

    private class ProcessRunnable implements Runnable {

        @Override
        public void run() {
            while (true) {
                try {
                    Thread.sleep(DEFAULT_SLEEP_TIME);
                    TaskManager.this.processTasks();
                } catch (InterruptedException e) {
                    log.error(e.toString(),e);
                }
            }
        }
    }
}
